package n80_自定义线程池;

import lombok.extern.slf4j.Slf4j;

import java.util.ArrayDeque;
import java.util.Deque;
import java.util.HashSet;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.ReentrantLock;

import static utils.Sleeper.sleep;

/**
 * @author fu
 * @date 2022-02-22 21:48
 * 概要：
 * --------------------------------------------------------------------------------
 * |                                                                              |
 * |       线程池：创建一批线程 得到重复的利用  减少内存占用 和 线程数量 ；避免上下文切换   |
 * |                                                                             |
 * |==============================================================================
 * 线程不能建太多
 * 多了 超过CPU核心，会造成 某些线程获取不到CPU时间片而阻塞； 造成上下文切换 而性能降低
 *
 * 应该充分利用 已有线程的潜力
 *
 */
@Slf4j(topic = "c.Test")
public class T1_take死等与poll超时_and_当任务队列已满 {

    public static void main(String[] args) throws InterruptedException {

        final ThreadPool threadPool = new ThreadPool(2, 1000, TimeUnit.MILLISECONDS, 10);

        // 主线程 提交任务  --- 注意 i 是变化的 不能被 lamada 直接使用 所以提出来 做变量
//        for (int i = 0; i < 5; i++) {
//
//            int j = i;
//            threadPool.execute(()->{
//
//                log.debug("{}", j);
//
//            });
//        }

//=======================================================================================================
        for (int i = 0; i < 15; i++) {   // 15个任务， 任务队列10+线程池可执行2  【第十三个线程 [ put 会阻塞]】

            int j = i;
            threadPool.execute(()->{
                sleep(20000);  // 执行时间拉长 只看 任务队列满的情况
                log.debug("{}", j);

            });
        }

    }
}

// 线程池 类
@Slf4j(topic = "c.Pool")
class ThreadPool {

    // 任务队列
    private BlockingQueue<Runnable> taskQueue; // 【用来放入 任务】

    // 线程集合
    private HashSet<Worker> workers = new HashSet(); // 这里有共享数据 在 execute()方法里用锁

    // 核心线程数
    private int coreSize;

    // 设置超时时间，过一段时间 还没任务 则让线程结束
    private long timeout;
    private TimeUnit timeUnit;

    public ThreadPool(int coreSize, long timeout, TimeUnit timeUnit, int queueCapcity) {
        this.coreSize = coreSize;
        this.timeout = timeout;
        this.timeUnit = timeUnit;
        this.taskQueue = new BlockingQueue<>(queueCapcity);
    }

    // 执行任务 ： 来一个任务，如果 [线程集合]里任务数 <  核心数 ; 放入 [线程集合] 并且执行
    //            否则 放入 任务队列
    public void execute(Runnable task){

        synchronized (workers){

            // 1. 任务数 <  核心数 coreSize ， 交给 worker 创建线程对象 & 来执行任务
            if(workers.size() < coreSize){
                final Worker worker = new Worker(task);
                log.debug("新增 worker {}, task {}", worker, task);
                workers.add(worker);

                worker.start();

                // 2. 如果超过 coreSize , 则加入 任务队列 暂存
            }else{

                taskQueue.put(task);// 这里满了会阻塞 ； 执行完了会唤醒 take阻塞的

            }
        }

    }

    // 线程包装
    class Worker extends Thread{

        private Runnable task;

        public Worker(Runnable task) {
            this.task = task;
        }

        /**
         * 要么从 初始化 里获得 任务执行
         * 要么从 任务队列里 获取 任务执行
         */
        @Override
        public void run() {

//HEAD>>>>>>>>>>>>>>>>>>>>>>>>>>>
//            while(task != null    ||    (task = taskQueue.take()) != null){// take() 会 【阻塞等待  没有任务时 线程不会停止】
//===============================
            while(task != null    ||    (task = taskQueue.poll(timeout,timeUnit)) != null){// take()  【没有任务时 超时停止】
//END<<<<<<<<<<<<<<<<<<<<<<<<<<<<
                try{
                    log.debug("正在执行...{}", task);
                    task.run();

                } catch (Exception e){

                } finally {
                    task = null;
                }
            }
            // 移除 任务
            synchronized (workers) {
                log.debug("worker 被移除 {}",this);
                workers.remove(this);
            }

        }
    }


}
// 测试，拿到锁的程序 是原子的 一个一个执行的，没拿到锁时等待的
@Slf4j(topic = "c.T")
class Test{
    private ReentrantLock lock = new ReentrantLock();
    private Condition fullWaitSet = lock.newCondition();

    public static void main(String[] args) {

        final Test t = new Test();
        new  Thread(()->{
            t.test1();
        }).start();
        new  Thread(()->{
            t.test2();
        }).start();


    }

    public void test1(){
        lock.lock();
        try {
            log.debug("1111 休息");
            sleep(5);
            log.debug("1111 醒来");
        } finally {
            lock.unlock();
        }

    }

    public void test2(){
        lock.lock();
        try {
            log.debug("2222 休息");
            sleep(5);
            log.debug("2222 醒来");
        } finally {
            lock.unlock();
        }
    }

}


// 任务队列 类
@Slf4j(topic = "c.Block")
class BlockingQueue<T> {

    // 1. 任务队列  使用双向链表
    private Deque<T> queue = new ArrayDeque<>(); // ArrayDeque 通常 比 LinkedList 性能更好

    // 2. 锁     将来线程池都要从 任务队列的 头部 获取任务 【取任务时 一个一个获取，前面线程获取 后面得等待】
    private ReentrantLock lock = new ReentrantLock();

    // 3. 针对生产者的 条件变量 【不可以无限量的 往消息队列里 添加任务】
    private Condition fullWaitSet = lock.newCondition();

    // 4. 针对消费者的 条件变量 【没有任务时(队列空)等待】
    private Condition emptyWaitSet = lock.newCondition();

    // 5. 容量
    private int capcity;

    public BlockingQueue(int capcity) {
        this.capcity = capcity;
    }

    // 阻塞获取 --- 带超时
    public T poll(long timeOut, TimeUnit unit){
        lock.lock();

        try{
            // 1. 转成 纳秒
            long nanos = unit.toNanos(timeOut);
            // 没有 元素 --- 阻塞等待
            while(queue.isEmpty()){
                try {
                    // 小于等于0 就是等待超时了
                    if(nanos <= 0){
                        log.debug("--- 就是没有了 超时了也没有了 ---");
                        return null;
                    }
                    nanos = emptyWaitSet.awaitNanos(nanos);// 返回 等待剩余时间，防止虚假唤醒 重复等待，而要让其等待剩余时间
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }

            // 有 元素 获取头部元素 返回
            final T t = queue.removeFirst();

            fullWaitSet.signal();// 【唤醒 , 可以往消息队列里 放入信息了】
            return t;

        } finally {

            lock.unlock();
        }

    }

    // 阻塞获取 --- 从队列中 获取元素
    public T take(){
        lock.lock();

        try{
            // 没有 元素 --- 阻塞等待
            while(queue.isEmpty()){
                try {
                    emptyWaitSet.await();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }

            // 有 元素 获取头部元素 返回
            final T t = queue.removeFirst();

            fullWaitSet.signal();// 【唤醒 , 可以往消息队列里 放入信息了】
            return t;

        } finally {

            lock.unlock();
        }

    }
//=====================================================================================================================
    // 阻塞添加 --- 从队列尾部 添加元素
    public void put(T task){
        lock.lock();

        try{
            // 队列 容量已满 --- 阻塞等待
            while(size()>=capcity){
                try {
                    log.debug("等待加入队列 task {}...", task);
                    fullWaitSet.await();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
            log.debug("加入队列 task {}", task);
            // 未满 则存入 ,并且 唤醒上面阻塞的
            queue.addLast(task);

            emptyWaitSet.signal();// 【唤醒 , 可以往消息队列里 拿出信息了】

        }finally {

            lock.unlock();
        }
    }
    // 阻塞添加 --- 从队列尾部 添加元素 【带超时时间】
    public boolean offer(T task, long timeOut, TimeUnit unit){
        lock.lock();

        try{
            // 1. 转成 纳秒
            long nanos = unit.toNanos(timeOut);

            // 队列 容量已满 --- 阻塞等待
            while(size()>=capcity){
                try {
                    if(nanos <= 0)return false;

                    log.debug("等待加入队列 task {}...", task);
                    nanos = fullWaitSet.awaitNanos(nanos);

                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
            log.debug("加入队列 task {}", task);
            // 未满 则存入 ,并且 唤醒上面阻塞的
            queue.addLast(task);

            emptyWaitSet.signal();// 【唤醒 , 可以往消息队列里 拿出信息了】

            return true;
        }finally {

            lock.unlock();
        }
    }

    // 获取队列 大小 --- 获取队列里排队任务还有几个
    public int size(){
        lock.lock();

        try{

            return queue.size();

        }finally {
            lock.unlock();
        }

    }



}



